package cnm.red5;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

import org.red5.server.api.scheduling.IScheduledJob;
import org.red5.server.api.scheduling.ISchedulingService;
import org.red5.server.api.stream.IBroadcastStream;
import org.red5.server.api.stream.IStreamAwareScopeHandler;
import org.red5.server.api.stream.IStreamListener;
import org.red5.server.api.stream.IStreamPacket;
import org.red5.server.api.stream.ResourceExistException;
import org.red5.server.api.stream.ResourceNotFoundException;
import org.red5.server.messaging.IFilter;
import org.red5.server.messaging.IMessage;
import org.red5.server.messaging.IMessageComponent;
import org.red5.server.messaging.IMessageInput;
import org.red5.server.messaging.IMessageOutput;
import org.red5.server.messaging.IPassive;
import org.red5.server.messaging.IPipe;
import org.red5.server.messaging.IPipeConnectionListener;
import org.red5.server.messaging.IProvider;
import org.red5.server.messaging.IPushableConsumer;
import org.red5.server.messaging.OOBControlMessage;
import org.red5.server.messaging.PipeConnectionEvent;
import org.red5.server.net.rtmp.event.AudioData;
import org.red5.server.net.rtmp.event.IRTMPEvent;
import org.red5.server.net.rtmp.event.VideoData;
import org.red5.server.stream.AbstractStream;
import org.red5.server.stream.IProviderService;
import org.red5.server.stream.ISeekableProvider;
import org.red5.server.stream.message.RTMPMessage;
import org.red5.server.stream.message.ResetMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * An implementation for server side stream.
 * 
 * @author The Red5 Project (red5@osflash.org)
 * @author Steven Gong (steven.gong@gmail.com)
 */
public class CnmServerStream extends AbstractStream implements IBroadcastStream,
		IFilter, IPushableConsumer, IPipeConnectionListener {
    /**
     * Logger
     */
    private static final Logger log = LoggerFactory.getLogger(CnmServerStream.class);

    /**
     * Enumeration for states
     */
    protected enum State {
		UNINIT, CLOSED, STOPPED, PLAYING, PAUSED
	}

    private static final long WAIT_THRESHOLD = 0;
    
    /**
     * Current state
     */
    protected State state;
    /**
     * Stream published name
     */
    protected String publishedName;
    /**
     * Message input
     */
	private IMessageInput msgIn;
    /**
     * Message output
     */
	private IMessageOutput msgOut;
    /**
     * Scheduling service
     */
	private ISchedulingService scheduler;
    /**
     * VOD scheduled job name
     */
	private String vodJobName;
    /**
     * VOD start timestamp
     */
	private long vodStartTS;
    /**
     * Server start timestamp
     */
	private long serverStartTS;
    /**
     * Next msg's timestamp
     */
	private long nextTS;
    /**
     * Next RTMP message
     */
	private RTMPMessage nextRTMPMessage;

	/** Listeners to get notified about received packets. */
	private Set<IStreamListener> listeners = new CopyOnWriteArraySet<IStreamListener>();
	
	private List<CnmPlayItem> playlist = new ArrayList<CnmPlayItem>();

	/** Constructs a new ServerStream. */
    public CnmServerStream(List<CnmPlayItem> playlist) {
		state = State.UNINIT;
		if (playlist == null) {
			throw new IllegalArgumentException("playlist should not be null!");
		}
		this.playlist = playlist;
	}

	/** {@inheritDoc} */
	public void saveAs(String name, boolean isAppend) throws IOException,
			ResourceNotFoundException, ResourceExistException {
		// save as not supported, do nothing
	}

	/** {@inheritDoc} */
	public String getSaveFilename() {
		return null;
	}

	/** {@inheritDoc} */
    public IProvider getProvider() {
		return this;
	}

	/** {@inheritDoc} */
    public String getPublishedName() {
		return publishedName;
	}

	/** {@inheritDoc} */
    public void setPublishedName(String name) {
		publishedName = name;
	}

	/**
	 * Start this server-side stream
	 */
	public void start() {
		if (state != State.UNINIT) {
			throw new IllegalStateException("State " + state
					+ " not valid to start");
		}
		if (publishedName == null) {
			throw new IllegalStateException(
					"A published name is needed to start");
		}
		// publish this server-side stream
		IProviderService providerService = (IProviderService) getScope()
				.getContext().getBean(IProviderService.BEAN_NAME);
		providerService
				.registerBroadcastStream(getScope(), publishedName, this);
		scheduler = (ISchedulingService) getScope().getContext().getBean(
				ISchedulingService.BEAN_NAME);
		state = State.STOPPED;
		for (CnmPlayItem item : playlist) {
			scheduler.addScheduledOnceJob(item.getStartDate(), new PlayJob(item));
		}
		
	}

    /**
     * Stop this server-side stream
     */
    public synchronized void stop() {
		if (state != State.PLAYING && state != State.PAUSED) {
			return;
		}
		if (vodJobName != null) {
			scheduler.removeScheduledJob(vodJobName);
			vodJobName = null;
		}
		if (msgIn != null) {
			msgIn.unsubscribe(this);
			msgIn = null;
		}
		if (nextRTMPMessage != null) {
			nextRTMPMessage.getBody().release();
		}
		state = State.STOPPED;
	}
	
	/** {@inheritDoc} */
    public synchronized void close() {
		if (state == State.PLAYING || state == State.PAUSED) {
			stop();
		}
		if (msgOut != null) {
			msgOut.unsubscribe(this);
		}
		notifyBroadcastClose();
		state = State.CLOSED;
	}

	/** {@inheritDoc} */
    public void onOOBControlMessage(IMessageComponent source, IPipe pipe,
			OOBControlMessage oobCtrlMsg) {
	}

	/** {@inheritDoc} */
    public void pushMessage(IPipe pipe, IMessage message) throws IOException {
		pushMessage(message);
	}


    /**
     * Pipe connection event handler. There are two types of pipe connection events so far,
     * provider push connection event and provider disconnection event.
     *
     * Pipe events handling is the most common way of working with pipes.
     *
     * @param event        Pipe connection event context
     */
    public void onPipeConnectionEvent(PipeConnectionEvent event) {
		switch (event.getType()) {
			case PipeConnectionEvent.PROVIDER_CONNECT_PUSH:
				if (event.getProvider() == this
						&& (event.getParamMap() == null || !event.getParamMap()
								.containsKey("record"))) {
					this.msgOut = (IMessageOutput) event.getSource();
				}
				break;
			case PipeConnectionEvent.PROVIDER_DISCONNECT:
				if (this.msgOut == event.getSource()) {
					this.msgOut = null;
				}
				break;
			default:
				break;
		}
	}

	/**
	 * Play a specific IPlayItem.
	 * The strategy for now is VOD first, Live second.
	 * Should be called in a synchronized context.
     *
	 * @param item        Item to play
	 */
	protected void play(CnmPlayItem item) {
        // Return if already playing
        if (state != State.STOPPED) {
			return;
		}
        // Get provider service from Spring bean factory
        IProviderService providerService = (IProviderService) getScope()
				.getContext().getBean(IProviderService.BEAN_NAME);
		msgIn = providerService.getVODProviderInput(getScope(), item.getName());
		if (msgIn == null) {
			// looks like a live stream, ignore
			return;
		}
		state = State.PLAYING;
		sendResetMessage();
		msgIn.subscribe(this, null);
		startBroadcastVOD();
	}

    /**
     * Play next item on item end
     */
	protected void onItemEnd() {
		stop();
	}

    /**
     * Push message
     * @param message     Message
     */
    private void pushMessage(IMessage message) throws IOException {
		msgOut.pushMessage(message);
		
		// Notify listeners about received packet
		if (message instanceof RTMPMessage) {
			final IRTMPEvent rtmpEvent = ((RTMPMessage) message).getBody();
			if (rtmpEvent instanceof IStreamPacket) {
				for (IStreamListener listener: getStreamListeners()) {
					try {
						listener.packetReceived(this, (IStreamPacket) rtmpEvent);
					} catch (Exception e) {
						log.error("Error while notifying listener " + listener, e);
					}
				}
			}
		}
	}

    /**
     * Send reset message
     */
    private void sendResetMessage() {
        // Send new reset message
    	try {
    		pushMessage(new ResetMessage());
    	} catch (IOException err) {
    		log.error("Error while sending reset message.", err);
    	}
	}

    /**
     * Begin VOD broadcasting
     */
    protected void startBroadcastVOD() {
		nextRTMPMessage = null;
		vodStartTS = 0;
		serverStartTS = System.currentTimeMillis();
		IStreamAwareScopeHandler handler = getStreamAwareHandler();
		if (handler != null) {
			handler.streamPublishStart(this);
		}
		notifyBroadcastStart();
		scheduleNextMessage();
	}

	/**
	 *  Notifies handler on stream broadcast stop
	 */
    protected void notifyBroadcastClose() {
		IStreamAwareScopeHandler handler = getStreamAwareHandler();
		if (handler != null) {
			try {
				handler.streamBroadcastClose(this);
			} catch (Throwable t) {
				log.error("error notify streamBroadcastStop", t);
			}
		}
	}

	/**
	 *  Notifies handler on stream broadcast start
	 */
    protected void notifyBroadcastStart() {
		IStreamAwareScopeHandler handler = getStreamAwareHandler();
		if (handler != null) {
			try {
				handler.streamBroadcastStart(this);
			} catch (Throwable t) {
				log.error("error notify streamBroadcastStart", t);
			}
		}
	}

	/**
	 * Pull the next message from IMessageInput and schedule
	 * it for push according to the timestamp.
	 */
	protected void scheduleNextMessage() {
		boolean first = nextRTMPMessage == null;
		long delta;
		
		while (true) {
			nextRTMPMessage = getNextRTMPMessage();
			if (nextRTMPMessage == null) {
				onItemEnd();
				return;
			}

			IRTMPEvent rtmpEvent = nextRTMPMessage.getBody();
			// filter all non-AV messages
			if (!(rtmpEvent instanceof VideoData)
					&& !(rtmpEvent instanceof AudioData)) {
				continue;
			}
			rtmpEvent = nextRTMPMessage.getBody();
			nextTS = rtmpEvent.getTimestamp();
			if (first) {
				vodStartTS = nextTS;
				first = false;
			}

			delta = nextTS - vodStartTS - (System.currentTimeMillis() - serverStartTS);
			if (delta < WAIT_THRESHOLD) {
				if (!doPushMessage()) {
					return;
				}
				if (state != State.PLAYING) {
					// Stream is paused, don't load more messages
					nextRTMPMessage = null;
					return;
				}
			} else {
				break;
			}
		}
		vodJobName = scheduler.addScheduledOnceJob(delta, new IScheduledJob() {
			/** {@inheritDoc} */
            public void execute(ISchedulingService service) {
				synchronized (CnmServerStream.this) {
					if (vodJobName == null) {
						return;
					}
					vodJobName = null;
					if (!doPushMessage()) {
						return;
					}
					if (state == State.PLAYING) {
						scheduleNextMessage();
					} else {
						// Stream is paused, don't load more messages
						nextRTMPMessage = null;
					}
				}
			}
		});
	}
	
	private boolean doPushMessage() {
		boolean sent = false;
		if (nextRTMPMessage != null) {
			sent = true;
			try {
				pushMessage(nextRTMPMessage);
	    	} catch (IOException err) {
	    		log.error("Error while sending message.", err);
	    	}
			nextRTMPMessage.getBody().release();
		}
		return sent;
	}

	/**
     * Getter for next RTMP message.
     *
     * @return  Next RTMP message
     */
    protected RTMPMessage getNextRTMPMessage() {
		IMessage message;
		do {
            // Pull message from message input object...
			try {
				message = msgIn.pullMessage();
			} catch (IOException err) {
				log.error("Error while pulling message.", err);
				message = null;
			}
            // If message is null then return null
            if (message == null) {
				return null;
			}
		} while (!(message instanceof RTMPMessage));
        // Cast and return
        return (RTMPMessage) message;
	}

    /**
     * Send VOD initialization control message
     * @param msgIn            Message input
     * @param start            Start timestamp
     */
    private void sendVODInitCM(IMessageInput msgIn, int start) {
        // Create new out-of-band control message
        OOBControlMessage oobCtrlMsg = new OOBControlMessage();
        // Set passive type
        oobCtrlMsg.setTarget(IPassive.KEY);
        // Set service name of init
        oobCtrlMsg.setServiceName("init");
        // Create map for parameters
        Map<Object, Object> paramMap = new HashMap<Object, Object>();
        // Put start timestamp into Map of params
        paramMap.put("startTS", start);
        // Attach to OOB control message and send it
        oobCtrlMsg.setServiceParamMap(paramMap);
		msgIn.sendOOBControlMessage(this, oobCtrlMsg);
	}

    /**
     * Send VOD seek control message
     * 
     * @param msgIn				Message input
     * @param position			New timestamp to play from
     */
	private void sendVODSeekCM(IMessageInput msgIn, int position) {
		OOBControlMessage oobCtrlMsg = new OOBControlMessage();
		oobCtrlMsg.setTarget(ISeekableProvider.KEY);
		oobCtrlMsg.setServiceName("seek");
		Map<Object, Object> paramMap = new HashMap<Object, Object>();
		paramMap.put("position", new Integer(position));
		oobCtrlMsg.setServiceParamMap(paramMap);
		msgIn.sendOOBControlMessage(this, oobCtrlMsg);
		synchronized (this) {
			// Reset properties
			vodStartTS = 0;
			serverStartTS = System.currentTimeMillis();
			if (nextRTMPMessage != null) {
				try {
					pushMessage(nextRTMPMessage);
		    	} catch (IOException err) {
		    		log.error("Error while sending message.", err);
		    	}
				nextRTMPMessage.getBody().release();
				nextRTMPMessage = null;
			}
			ResetMessage reset = new ResetMessage();
			try {
				pushMessage(reset);
	    	} catch (IOException err) {
	    		log.error("Error while sending message.", err);
	    	}
			scheduleNextMessage();
		}
	}

	public void addStreamListener(IStreamListener listener) {
		listeners.add(listener);
	}

	public Collection<IStreamListener> getStreamListeners() {
		return listeners;
	}

	public void removeStreamListener(IStreamListener listener) {
		listeners.remove(listener);
	}
	
	private class PlayJob implements IScheduledJob {
		private CnmPlayItem item;
		
		PlayJob(CnmPlayItem item) {
			this.item = item;
		}

		public void execute(ISchedulingService service)
				throws CloneNotSupportedException {
			synchronized (CnmServerStream.this) {
				if (state == State.PLAYING) {
					stop();
				}
				play(item);
			}
		}
	}
}